// Copyright 2012 Google Inc. All rights reserved.

package sqlite

import (
	"log"

	"time"

	tsdb "code.google.com/p/gomonitor/tsdb"
	sql "github.com/kuroneko/gosqlite3"
)

var (
	schemas = [][]string{
		{ // Init: Schema base
			"CREATE TABLE IF NOT EXISTS series (scope TEXT, metric TEXT NOT NULL, " +
				"value FLOAT, ts INTEGER NOT NULL)",
		},
	}
	selectSql = "SELECT value, ts FROM series WHERE scope = ? AND metric = ? AND (ts >= ? AND ts <= ?) ORDER BY ts ASC"
	tickerSql = "SELECT value, max(ts), FROM series WHERE scope = ? AND metric = ?"
	insertSql = "INSERT OR IGNORE INTO series (scope, metric, value, ts) VALUES (?, ?, ?, ?)"
)

type sqlDb struct {
	file string
}

// New conscructs a sqlite3 implementation of the TsDb interface.
// it returns the db if successful or an error if not.
func New(file string) (db tsdb.TsDb, err error) {
	// TODO(jwall): Make this based off of a db version
	sql.Session(file, func(db *sql.Database) {
		err = db.Begin()
		if err != nil {
			return
		}
		for _, sts := range schemas {
			for _, s := range sts {
				_, err = db.Execute(s)
				if err != nil {
					return
				}
			}
		}
		err = db.Commit()
	})
	db = &sqlDb{file: file}
	return
}

func metricKey(m tsdb.Metric) string {
	return m.Scope + m.Var
}

func dpFactory(ts time.Time, sz int) tsdb.Datapoint {
	return tsdb.Datapoint{Ts: ts, Value: make([]*float64, sz)}
}

func (db *sqlDb) Ticker(metrics []tsdb.Metric, d time.Duration) (chan<- bool, <-chan *tsdb.SeriesResponse) {
	ch := make(chan *tsdb.SeriesResponse)
	stop := make(chan bool)
	tCh := time.NewTicker(d)
	defer tCh.Stop()
	go func() {
		defer close(ch)
		for {
			select {
			case <-tCh.C:
				go sql.Session(db.file, func(db *sql.Database) {
					vals := []tsdb.Datapoint{}
					for i, metric := range metrics {
						st, err := db.Prepare(tickerSql, metric.Scope, metric.Var)
						if err != nil {
							log.Printf("SqlError: %q", err)
							ch <- &tsdb.SeriesResponse{E: err}
							return
						}
						e := st.Step(func(s *sql.Statement, cols ...interface{}) {
							fv := cols[0].(float64)
							nano := cols[1].(int64)
							sec := nano / int64(time.Second)
							nsec := nano % int64(time.Second)
							dp := dpFactory(time.Unix(sec, nsec), len(metrics))
							dp.Value[i] = &fv
							vals = append(vals, dp)
						})
						if e != nil {
							log.Printf("Error retreiving value: %q", e)
						}
						st.Reset()
					}
					for _, r := range tsdb.RespFactory(vals, len(metrics)) {
						ch <- &r
					}
				})
			case <-stop:
				return
			}
		}
	}()
	return stop, ch
}

func (db *sqlDb) GetMultipleSeries(metrics []tsdb.Metric, end time.Time, d time.Duration) []tsdb.SeriesResponse {
	log.Printf("Asked time %q", end)
	end = end.UTC()
	log.Printf("UTC time %q", end)
	start := end.Add(-d)
	vals := []tsdb.Datapoint{}
	sql.Session(db.file, func(db *sql.Database) {
		sts := []*sql.Statement{}
		for _, metric := range metrics {
			log.Printf("Getting series between %s and %s for %v",
				start, end, metric)
			st, err := db.Prepare(
				selectSql, metric.Scope, metric.Var, start.UnixNano(), end.UnixNano())
			if err != nil {
				log.Printf("SqlError: %q", err)
				// TODO(jwall): Set error for return here.
				return
			}
			sts = append(sts, st)
		}
		for {
			sts2 := sts
			if len(sts) == 0 {
				break
			}
			for i, st := range sts2 {
				e := st.Step(func(s *sql.Statement, cols ...interface{}) {
					//log.Printf("Retrieving a row for i'th st: %d", i)
					if len(cols) == 2 {
						nano := cols[1].(int64)
						sec := nano / int64(time.Second)
						nsec := nano % int64(time.Second)
						dp := dpFactory(time.Unix(sec, nsec), len(sts2))
						//log.Printf("dp.Value len %d", len(dp.Value))
						v := cols[0].(float64)
						dp.Value[i] = &v
						vals = append(vals, dp)
						//log.Printf("Adding %#v to the series", *(dp.Value[0]))
					} else {
						log.Printf("Unexpected length for %v", cols)
					}
				})
				if e != sql.ROW {
					//log.Printf("Exhausted statement %q %v", st.SQLSource(), e)
					//log.Printf("sts len %d index %d", len(sts), i)
					st.Reset()
					// remove the element from sts
					if len(sts) == 1 {
						sts = nil
						break
					}
					if i < (len(sts) - 1) {
						sts = append(sts[:i], sts[i+1:]...)
					} else {
						sts = sts[:i]
					}
				}
			}
		}
	})
	return tsdb.RespFactory(vals, len(metrics))
}

func (db *sqlDb) Add(metric tsdb.Metric, value float64, ts time.Time) (err error) {
	ts = ts.UTC()
	sql.Session(db.file, func(db *sql.Database) {
		//log.Printf("Adding value %.2f for %q at t=%q", value, metric, ts)
		var st *sql.Statement
		st, err = db.Prepare(insertSql, metric.Scope, metric.Var, value, ts.UnixNano())
		if err != nil {
			return
		}
		_, err = st.All()
	})
	return
}
